package com.eeboot;

import com.alibaba.fastjson.JSON;
import com.eeboot.vo.Log;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class KafkaInputTransform {
    public static void main(String[] args) throws Exception{
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.19.80.176:9092");
        props.put("zookeeper.connect", "172.19.80.176:2181");
        props.put("group.id", "flink-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");

        String topic = "filebeat_bsd_bsd_service";

        DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer011<>(
                topic,
                new SimpleStringSchema(),
                props
        )).setParallelism(1);

        SingleOutputStreamOperator<Log> regMap = dataStreamSource.map(new LogMapFucntion());

        SingleOutputStreamOperator<Tuple2<String,Integer>> profileMap = regMap.map(new MapFunction<Log, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Log value) throws Exception {
                return new Tuple2<>(value.getProfile(),1);
            }
        });

        SingleOutputStreamOperator<Tuple2<String,Integer>> countStream = profileMap.keyBy(0).timeWindow(Time.minutes(1))
                .sum(1);

        SingleOutputStreamOperator<Tuple2<String,Integer>> ipMap = regMap.map(new MapFunction<Log, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Log value) throws Exception {
                return new Tuple2<>(value.getIp(),1);
            }
        });

        SingleOutputStreamOperator<Tuple2<String,Integer>> countStream2 = ipMap.keyBy(0).timeWindow(Time.seconds(30))
                .sum(1);

        SingleOutputStreamOperator<Tuple2<String,Integer>> theadMap = regMap.map(new MapFunction<Log, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Log value) throws Exception {
                return new Tuple2<>(value.getLogThread().substring(0,value.getLogThread().indexOf("-")),1);
            }
        });

        SingleOutputStreamOperator<Tuple2<String,Integer>> countStream3 = theadMap.keyBy(0).timeWindow(Time.seconds(30))
                .sum(1);

        countStream.print();
        countStream2.print();
        countStream3.print();
        env.execute("kafka source");
    }

}


class LogMapFucntion implements MapFunction<String, Log>{

    private static final String DEFAULT_LOG_PATTERN = "\\[([^]]*)]\\s+\\[([^]]*)]\\s+\\[([^]]*)]\\s+\\[([^]]*)]\\s+(\\d{4}-\\d{2}-\\d{2}\\s\\d{2}:\\d{2}:\\d{2}\\.\\d{3})\\s+\\[([^]]*)]\\s+\\[([^]]*)]\\s+\\[([^]]*)]\\s+(\\w{3,5})\\s+(.*)";
    Pattern pattern = null;

    public LogMapFucntion() {
        pattern = Pattern.compile(DEFAULT_LOG_PATTERN);
    }

    @Override
    public Log map(String value) throws Exception {
        String message = JSON.parseObject(value).getString("message");
        Log log = new Log();
        //log.setMessage(message);
        Matcher matcher = pattern.matcher(message);
        if (matcher.find()) {
            if(matcher.groupCount()==10){
                log.setApp(String.valueOf(matcher.group(1)));
                log.setProfile(String.valueOf(matcher.group(2)));
                log.setHostName(String.valueOf(matcher.group(3)));
                log.setIp(String.valueOf(matcher.group(4)));
                log.setLogTime(String.valueOf(matcher.group(5)));
                log.setLogThread(String.valueOf(matcher.group(6)));
                log.setTid(String.valueOf(matcher.group(7)));
                log.setSpid(String.valueOf(matcher.group(8)));
                log.setLogLevel(String.valueOf(matcher.group(9)));
                log.setLogData(String.valueOf(matcher.group(10)));
            }
        }

        return log;
    }
}